package com.example.actordemo1.routering;

import akka.actor.AbstractActor;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.routing.ConsistentHashingPool;
import akka.routing.ConsistentHashingRouter;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;

public class HashActor  extends AbstractActor {

    Map<String, String> cache = new HashMap<String, String>();
    public static final String NOT_FOUND = "NOT_FOUND";
    @Override
    public Receive createReceive() {
        return receiveBuilder()
                .match(
                        Entry.class,
                        entry -> {
                            System.out.println("entry"+entry);
                            cache.put(entry.key, entry.value);
                        })
                .match(
                        Get.class,
                        get -> {
                            System.out.println("get"+get.key);
                            Object value = cache.get(get.key);
                            getSender().tell(value == null ? NOT_FOUND : value, getSelf());
                        })
                .match(
                       Evict.class,
                        evict -> {
                            System.out.println("evict");
                            cache.remove(evict.key);
                        })
                .build();
    }

    public static void main(String[] args) {

        final ConsistentHashingRouter.ConsistentHashMapper  hashMapper = (Object msg)->{
            if(msg instanceof Evict){
                return ((Evict)msg).key;
            }else{
                return  null;
            }

        };
//看到了Get消息ConsistentHashable本身的实现，
// 而Entry消息则包装在中ConsistentHashableEnvelope。
// 该Evict消息由hashMapping部分功能处理。
        ActorSystem actorSystem = ActorSystem.create("sys");
        ActorRef actorRef = actorSystem.actorOf(new ConsistentHashingPool(10)
                .withHashMapper(hashMapper).props(Props.create(HashActor.class)),"hashActor");

        actorRef.tell(new ConsistentHashingRouter
                .ConsistentHashableEnvelope(new Entry("hello", "HELLO"), "hello"), ActorRef.noSender());
//        actorRef.tell(new ConsistentHashingRouter
//                .ConsistentHashableEnvelope(new Entry("hi", "HI"), "hi"), ActorRef.noSender());
        actorRef.tell(new Get("hello"), ActorRef.noSender());
//        actorRef.tell(new Get("hello"), ActorRef.noSender());
//
//        actorRef.tell(new Get("hi"), ActorRef.noSender());
//
        actorRef.tell(new Evict("hi"),ActorRef.noSender());
//        actorRef.tell(new Get("hi"),ActorRef.noSender());

    }
}

 class Evict implements Serializable {
    private static final long serialVersionUID = 1L;
    public final String key;

    public Evict(String key) {
        this.key = key;
    }
}

// #cache-actor

// #cache-actor
 class Get implements Serializable, ConsistentHashingRouter.ConsistentHashable {
    private static final long serialVersionUID = 1L;
    public final String key;

    public Get(String key) {
        this.key = key;
    }

    public Object consistentHashKey() {
        return key;
    }
}

 class Entry implements Serializable {
    private static final long serialVersionUID = 1L;
    public final String key;
    public final String value;

    public Entry(String key, String value) {
        this.key = key;
        this.value = value;
    }

     @Override
     public String toString() {
         return "Entry{" +
                 "key='" + key + '\'' +
                 ", value='" + value + '\'' +
                 '}';
     }
 }
